package cn.com.gf.cmp.am.monitor.entrust;

import cn.com.gf.cmp.am.message.MessageTransform;
import cn.com.gf.cmp.am.message.ResponseMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 撤单数据推送：解决撤单数量为o问题。
 * 原来：推送撤单数据
 * 现在：推送委托数据中撤单数量字段不为0的数据
 * Created by zzh on 2017/11/30.
 */
@Component
public class CancelEntrustSourceMonitor {

    private final Logger logger= LoggerFactory.getLogger(CancelEntrustSourceMonitor.class);

    @Resource
    private JdbcTemplate jdbcTemplate;

    @Resource(name="entrustKafkaTemplate")
    private KafkaTemplate<String,String> template;

    @Value("${spring.kafka.amentrust.topics}")
    private String topic ;

    @Value("${table.entrust}")
    private String entrustTable="tentrusts";

    @Value("${table.cancelentrust.backup}")
    private String cancelEntrustTable="tcancel_entrust_kafka";

    @Value("${stop.time}")
    private int stopTime=153000;

    @Value("${persist.batch.size}")
    private int persistBatchSize=5000;

    private String SELECT_SQL="";
    private String INSERT_SQL="";
    private String DELETE_SQL="";

    private ObjectMapper mapper=new ObjectMapper();

    //调度间隔频率：15分钟
    @Scheduled(fixedDelay = 20*60*1000)
    public void loadAndProduce(){

        try{

        String branchNo="";
        String ldate="";
        String serialNo="";
        SELECT_SQL = "SELECT 'gw_cancel_entrust' as message_type,'8815' as branch_no, entrust.* FROM "+entrustTable+" entrust WHERE withdrawqty<>0 and ordertype not like '%C%' and not exists(select 1 from tcancel_entrust_kafka ka where substr(entrust.offertime,0,8)=ka.l_date and '8815'=ka.branch_no and entrust.serialnum=ka.serial_no) order by offertime asc";
        INSERT_SQL="insert into "+cancelEntrustTable+"(l_date,branch_no,serial_no) values(?,?,?)";
        DELETE_SQL="delete from  "+cancelEntrustTable+"  where l_date<"+DateTime.now().toString("yyyyMMdd");


         DateTime now= DateTime.now();
         DateTime stopDateTime=DateTime.parse(String.format("%06d",stopTime), DateTimeFormat.forPattern("HHmmss")).withDate(now.getYear(),now.getMonthOfYear(),now.getDayOfMonth());

        if(now.isBefore(stopDateTime)){
            jdbcTemplate.execute(DELETE_SQL);
        }

        List<Map<String,Object>> datalist=jdbcTemplate.queryForList(SELECT_SQL);
        List<Object[]> insertList=new ArrayList<>();
        MessageTransform tran =new MessageTransform();
        for(Map<String,Object> data:datalist){
            String stock_account=data.get("regid").toString();
            ResponseMessage message=tran.responseCancelTransform(data);
            //template.send(topic,stock_account,mapper.writeValueAsString(message));
            logger.debug(mapper.writeValueAsString(message));

            ldate=data.get("offertime").toString().substring(0,8);;
            branchNo=data.get("branch_no").toString();
            serialNo=data.get("serialnum").toString();
            insertList.add(new Object[]{ldate,branchNo,serialNo});

            if(insertList.size()==persistBatchSize){
                jdbcTemplate.batchUpdate(INSERT_SQL,insertList);
                insertList.clear();
            }
        }
        if(insertList!=null && !insertList.isEmpty()){
            jdbcTemplate.batchUpdate(INSERT_SQL,insertList);
            insertList.clear();
        }

        }catch(Exception e){
            logger.error(e.getMessage(),e);
        }
    }
}
